#include "Subscriber.h"
#include "Broker.h"
#include <zmq.h>
#include <assert.h>
#include <unistd.h>
#include <string.h>
#include <algorithm>

Subscriber::Subscriber():sub_(0),map_()
{
  sub_ = zmq_socket(Broker::Context(), ZMQ_SUB);
  assert(sub_);

  const int rc=zmq_connect(sub_, Broker::XpubAddr);
  assert(rc==0);
  start();
}

Subscriber::~Subscriber()
{
  stop();
  zmq_close(sub_);
  unregister();
}

void Subscriber::registerHandler(::google::protobuf::Message& msg, DataReader& reader)
{
  std::string topic = msg.GetTypeName();
  const int rc = zmq_setsockopt(sub_, ZMQ_SUBSCRIBE, topic.c_str(), topic.length());
  assert(rc == 0);
  map_[topic].push_back(&reader);
}

void Subscriber::unregisterHandler(::google::protobuf::Message& msg, DataReader& reader)
{
  std::string topic = msg.GetTypeName();
  ReaderList& L=map_[topic];
  L.erase(std::remove(L.begin(),L.end(),&reader),L.end());
}

void Subscriber::run(void)
{
  while(1)
  {
    // grab topic
    std::string topic;
    {
      zmq_msg_t msg;
      const int len=4096;
      const int rc0 = zmq_msg_init_size(&msg, len);
      assert(rc0==0);
      const int rc1=zmq_msg_recv(&msg,sub_,0);
      assert(rc1 > 0);
      topic=std::string((char*)zmq_msg_data(&msg));
printf("(%s:%d) topic: '%s'\n",__FILE__,__LINE__,(char*)zmq_msg_data(&msg));
      zmq_msg_close (&msg);
    }

    // grab content
    {
      zmq_msg_t msg;
      const int len=4096;
      const int rc2 = zmq_msg_init(&msg);
      assert(rc2==0);
      const int rc3=zmq_msg_recv(&msg,sub_,0);
      assert(rc3 > 0);
char* S=(char*)zmq_msg_data(&msg);
S[rc3-1]='\0';

      // locate readers
      Map::iterator iter=map_.find(topic);
      if(iter != map_.end())
      {
        ReaderList L=iter->second;
        for (ReaderList::iterator iter = L.begin() ; iter != L.end(); ++iter)
        {
          DataReader* reader=*iter;
          reader->incomingMessage(zmq_msg_data(&msg), rc3);
        }
        zmq_msg_close (&msg);
      }
    }
  }
}

  void Subscriber::unregister()
  {
    map_.clear();
  }
